MQTTXをつかってMQTTでサクッと負荷をかけてみる

MQTTXのbenchコマンドを使ってみました。簡単にMQTTで負荷をかけることができます。

製造業

#MQTT

kimura.kosuke

2024.12.11

こんにちは、製造ビジネステクノロジー部の木村です。

製造業アドベントカレンダーの12/11 (水)の記事を投稿します。
https://adventar.org/calendars/10479

サーバーレスの難しいところ

製造ビジネステクノロジー部では前身のCX事業部時代も併せて多くのIoT案件を取り扱ってまいりました。
その中でもサーバーレスを中心とした技術選定をしたものが数多くあります。

https://classmethod.jp/cases/lixil/

サーバーレスのメリットを活かすことで素早く開発を進めることができる一方で、
数多くのサービスを組み合わせることから、以下のような点はサーバーレス特有の難しさなのではないかと思います。

サービスごとの連携が不適切で、パフォーマンスが出なかったり、スロットリングしやすい構成になってしまうことがある
コストの見積もりが難しい。仮に算出しても自信が持てない
クオータの管理が大変。中には上限緩和の手続きが難航したり、そもそもできないものもある

負荷試験をサクッと実施したい

さて、これらの問題点を早期に発見するために有効なのが負荷試験です。
ただ、機能開発で手一杯であったり、実施のハードルが高く感じられたりして、負荷試験は後回しにされがちです。

負荷試験の種類

負荷試験には種類があります。
参考: GrafanaのLoad testingの解説ページ

DeepLによる訳

スモークテスト：最小限の負荷でシステムが機能することを確認します。

平均的な負荷テスト：標準的なトラフィックでシステムがどのように機能するかを確認します。

ストレステスト：ピーク時のトラフィックでシステムがどのように機能するかを確認します

スパイクテスト：急激なトラフィックの増加に対してシステムがどのように機能するかを確認します。突発的なトラフィックの増加に対してシステムがどのように機能するかを確認します。

ブレークポイントテスト：トラフィックを徐々に増加させ、システムのブレークポイントを発見します。トラフィックを徐々に増加させ、システムの限界点を発見する。

ソークテスト：長時間負荷がかかった場合にシステムが劣化するかどうかを検出します。

ストレステストより上位の本格的な負荷テストについては、攻撃ツール側にも相応の性能が求められるので、環境の準備が必要で手間がかかります。
また、負荷の大きさによっては事前にAWSへの連絡が必要だったりと手続き面でも何かとハードルがあります。(※ リンクはEC2が対象ですが、他のサービスを使用する場合でも何らかの形で事前に連絡しておくのがベターかなと思います)

一方で、冒頭で触れた難しい点に目処をつけることを目的にするならば、スモークテストまたは平均的な負荷テストを早い段階で行うことができれば、軌道修正も容易なのではないかと思います。

負荷試験のツール

HTTPによる負荷テストについては、多くの選択肢があります。

一方、IoTでよく使用されるMQTTにおいては、あまり選択肢は多くありません。

ECS on Fargate環境でpaho mqtt clientを使用(参考: MQTTの負荷テストもバッチリ！！Locustを活用した分散負荷テスト環境の構築)
mqtt-jmeter

いずれもインフラの準備など少し手間がかかりそうです。

今回紹介したいのは、表題にもあるMQTTXです。

MQTTXとは

MQTTXはオープンソースのMQTT Clientです。
デスクトップアプリ、CLI、Webアプリが提供されており、macOS、Linux、windowsで動作します。

詳しい使い方は以下で紹介されています
(参考: 独自スクリプトも実行できる！モダンでクールな MQTT クライアント「MQTT X」の紹介)

その中でも今回使用するのはCLIの中のbenchコマンドです

やってみた

主要なツールのバージョン

名前	バージョン
aws-cdk	2.171.1
aws-cli	2.15.53
mqttx	1.11.0

iot ruleとlambdaを作成

CDKで作成します。
IoT RuleとLambdaを作成します。

import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as iot from 'aws-cdk-lib/aws-iot';
import * as lambdaNodejs from 'aws-cdk-lib/aws-lambda-nodejs';
import * as lambda from 'aws-cdk-lib/aws-lambda';

export class AppStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    const sampleFunction = new lambdaNodejs.NodejsFunction(this, 'SampleFunction', {
      runtime: lambda.Runtime.NODEJS_22_X,
      entry: 'functions/handler.ts',
      handler: 'handler',
    });

    const topicRule = new iot.CfnTopicRule(this, 'SampleTopicRule', {
      ruleName: 'SampleRule',
      topicRulePayload: {
        actions: [
          {
            lambda: {
              functionArn: sampleFunction.functionArn,
            },
          },
        ],
        sql: 'SELECT *, topic() as topic FROM "topic/test/+"',
      },
    });

    sampleFunction.addPermission('IoTRuleInvocation', {
      principal: new cdk.aws_iam.ServicePrincipal('iot.amazonaws.com'),
      action: 'lambda:InvokeFunction',
      sourceArn: topicRule.attrArn
    });
  }
}

デバイス証明書・秘密鍵・CA証明書の準備

thingの作成

$ export THING_NAME=sample-thing-20241204
$ aws iot create-thing --thing-name sample-thing-20241204
{
    "thingName": "sample-thing-20241204",
    "thingArn": "arn:aws:iot:ap-northeast-1:372082753702:thing/sample-thing-20241204",
    "thingId": "2fa86daf-b2a7-47d7-94f2-1599bcf68bca"
}

証明書の作成

$ aws iot create-keys-and-certificate --set-as-active > keys_and_certificate.json

鍵類作成

$ jq -r '.keyPair.PrivateKey' keys_and_certificate.json > private-key.pem
$ jq -r '.keyPair.PublicKey' keys_and_certificate.json > public-key.pem
$ jq -r '.certificatePem' keys_and_certificate.json > device-cert.pem

環境変数

$ export CERTIFICATE_ARN=$(jq -r '.certificateArn' keys_and_certificate.json)

証明書とthingをattach

$ aws iot attach-thing-principal --thing-name $THING_NAME --principal $CERTIFICATE_ARN

ポリシー作成

$ export POLICY_NAME=sample-policy-20241204
$ aws iot create-policy --policy-name $POLICY_NAME --policy-document '{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:Publish",
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": "iot:Connect",
      "Resource": "*"
    }
  ]
}'

ポリシーアタッチ

$ aws iot attach-policy --policy-name $POLICY_NAME --target $CERTIFICATE_ARN

ルートCA取得

$ curl https://www.amazontrust.com/repository/AmazonRootCA1.pem -o AmazonRootCA1.pem

コマンド実行

シンプルにpublishする

mqttx bench pub \
--topic 'topic/test/sample-client' \
--message '{"msg": "hello from cli"}' \
--count 1000 \
--limit 0 \
--message-interval 100 \
--interval 10 \
--hostname 'xxxxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com' \
--port 8883 \
--key private-key.pem \
--cert device-cert.pem \
--ca AmazonRootCA1.pem

各種オプションの解説はドキュメントをご覧ください。

結果は以下のように表示されます。

❯  Starting publish benchmark, connections: 1000, req interval: 10ms, message interval: 100ms
✔  [10/10] - Connected
✔  Created 1000 connections in 4.0s
Published total: 32977, message rate: 98/s

publish内容を動的に変更したい時

topicを動的に変更する

--topic 'topic/test/%u'

といった形で変数を埋め込むことが可能です。

使用できる変数は以下のとおりです

%u: username
%i: index
%u: client

message用のファイルを作成する

以下のようなファイルを作成する

payload.jsonl

{"msg": "no.1"}
{"msg": "no.2"}
{"msg": "no.3"}

コマンドを実行する

mqttx bench pub \
--topic 'topic/test/sample-client' \
--file-read payload.jsonl \
--split '\n' \
--count 4 \
--limit 0 \
--message-interval 100 \
--interval 10 \
--hostname 'xxxxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com' \
--port 8883 \
--key private-key.pem \
--cert device-cert.pem \
--ca AmazonRootCA1.pem